package cn.tiger.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.logging.LogChannel;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.core.plugins.PluginFolder;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;


import java.io.InputStream;
import java.util.*;

@Slf4j
public class KettleUtils {

    /**
     * resource目录路径
     */
    public static final String KETTLE_SOURCES = "kettleSources/";


    /**
     * 执行转换
     * @param initKettleParam
     * @param ktrFilePath 本地路径 如/usr/local/ktrsource/xxx.ktr、d:\\ktrsource\\xxx.ktr
     * @param kettleHome kettle安装目录（加载插件必传）
     * @return
     */
    public static boolean runKettleTransferByFilePath(Map<String, String> initKettleParam, String ktrFilePath,String kettleHome) {
        Trans trans = null;
        String uuid = UUID.randomUUID().toString();
        log.info("KettleUtils@runKettleTransferByFilePath [{}]:ktrFilePath={},kettleHome={}", uuid, ktrFilePath, kettleHome);
        try {
            // 加载出Kettle核心相关插件
            if (StringUtils.isNotEmpty(kettleHome)) {
                StepPluginType
                        .getInstance()
                        .getPluginFolders()
                        .add(new PluginFolder(kettleHome, false, true));
//			//支持大数据相关的插件
//			//支持kafka
//			StepPluginType.getInstance().getPluginFolders()
//					.add(new PluginFolder(kettleHome+
//							"\\system\\karaf\\system\\pentaho\\pentaho-big-data-kettle-plugins-kafka",
//							false, true));
            }
            // 初始化
            KettleEnvironment.init();
            EnvUtil.environmentInit();
            TransMeta transMeta = new TransMeta(ktrFilePath);
            transMeta.setLogLevel(LogLevel.DETAILED);
            // 转换
            trans = new Trans(transMeta);
            // 初始化trans参数，脚本中获取参数值：${variableName}
            if (initKettleParam != null) {
                for (String variableName : initKettleParam.keySet()) {
                    trans.setVariable(variableName, initKettleParam.get(variableName));
                }
            }

            //设置详细日志输出
            trans.setLogLevel(LogLevel.DETAILED);

            // 执行转换
            trans.execute(null);

            // 等待转换执行结束
            trans.waitUntilFinished();
            if (trans.getErrors() > 0) {
                log.info("KettleUtils@runKettleTransferByFilePath:" + uuid + " 执行失败");
                throw new Exception("KettleUtils@runKettleTransferByFilePath:" + uuid + " 执行失败");
            } else {
                log.info("KettleUtils@runKettleTransferByFilePath:" + uuid + " 执行成功");
            }
            return true;
        } catch (Exception e) {
            log.error("KettleUtils@runKettleTransferByFilePath:" + uuid, e);
            return false;
        }
    }

    /**
     * 执行转换
     * @param initKettleParam
     * @param ktrFilePath resource目录下ktr文件路径
     * @param kettleHome kettle安装目录（加载插件必传）
     * @return
     */
    public static boolean runKettleTransferByInputStream(Map<String, String> initKettleParam, String ktrFilePath,String kettleHome) {
        Trans trans = null;
        String uuid = UUID.randomUUID().toString();
        log.info("KettleUtils@runKettleTransferByInputStream [{}]:ktrFilePath={},kettleHome={}", uuid, ktrFilePath, kettleHome);
        try {
            // 加载出Kettle核心相关插件
            if (StringUtils.isNotEmpty(kettleHome)) {
                StepPluginType
                        .getInstance()
                        .getPluginFolders()
                        .add(new PluginFolder(kettleHome, false, true));
//			//支持大数据相关的插件
//			//支持kafka
//			StepPluginType.getInstance().getPluginFolders()
//					.add(new PluginFolder(kettleHome+
//							"\\system\\karaf\\system\\pentaho\\pentaho-big-data-kettle-plugins-kafka",
//							false, true));
            }
            // 初始化
            KettleEnvironment.init();
            EnvUtil.environmentInit();
            InputStream resourceAsInputStream = ResourceUtil.getResourceAsInputStream(KETTLE_SOURCES + ktrFilePath);
            log.info("KettleUtils@runKettleTransferByInputStream:" + uuid + " {resourceAsInputStream:" + resourceAsInputStream + "}");
            TransMeta transMeta = new TransMeta(resourceAsInputStream, null, true, null, null);
            transMeta.setLogLevel(LogLevel.DETAILED);
            // 转换
            trans = new Trans(transMeta);
            // 初始化trans参数，脚本中获取参数值：${variableName}
            if (initKettleParam != null) {
                for (String variableName : initKettleParam.keySet()) {
                    trans.setVariable(variableName, initKettleParam.get(variableName));
                }
            }

            //设置详细日志输出
            trans.setLogLevel(LogLevel.DETAILED);

            // 执行转换
            trans.execute(null);

            // 等待转换执行结束
            trans.waitUntilFinished();
            if (trans.getErrors() > 0) {
                log.info("KettleUtils@runKettleTransferByInputStream:" + uuid + " 执行失败");
                throw new Exception("KettleUtils@runKettleTransferByInputStream:" + uuid + " 执行失败");
            } else {
                log.info("KettleUtils@runKettleTransferByInputStream:" + uuid + " 执行成功");
            }
            return true;
        } catch (Exception e) {
            log.error("KettleUtils@runKettleTransferByInputStream:" + uuid, e);
            return false;
        }
    }

    /**
     * 执行作业
     *
     * @param initKettleParam
     * @param kjbFilePath
     * @return
     */
    public static boolean runKettleJob(Map<String, String> initKettleParam, String kjbFilePath) {
        String uuid = UUID.randomUUID().toString();
        log.info("KettleUtils@runKettleJob:" + uuid + " {kjbFilePath:" + kjbFilePath + "}");
        try {
            KettleEnvironment.init();
            // 初始化job路径
            JobMeta jobMeta = new JobMeta(kjbFilePath, null);
            Job job = new Job(null, jobMeta);
            // 初始化job参数，脚本中获取参数值：${variableName}
            if (initKettleParam != null) {
                for (String variableName : initKettleParam.keySet()) {
                    job.setVariable(variableName, initKettleParam.get(variableName));
                }
            }

            job.start();
            job.waitUntilFinished();
            if (job.getErrors() > 0) {
                log.info("KettleUtils@runKettleJob:" + uuid + " 执行失败");
            } else {
                log.info("KettleUtils@runKettleJob:" + uuid + " 执行成功");
            }
            return true;
        } catch (Exception e) {
            log.error("KettleUtils@runKettleJob:" + uuid, e);
            return false;
        }
    }

    public static void main(String[] args) {
        HashMap<String, String> param = new HashMap<>();
        param.put("where", "where id = 290");
//        param.put("where", "where 1=1");
//        String ktrPath = "C:\\Users\\WT\\Desktop\\3.1 pc_project_team_mysql2dat_20230506.ktr";
//        String ktrPath = "C:\\Users\\WT\\Desktop\\111.ktr";
//        boolean b = KettleUtils.runKettleTransferByFilePath(param, ktrPath,null);

        String inputKtrPath = "111.ktr";
        boolean b = KettleUtils.runKettleTransferByInputStream(param, inputKtrPath, null);
        System.out.println("b = " + b);
    }

}
